package algorithm

import org.apache.spark.graphx.{Graph, VertexId}

object MinSpanningTreeAlgorithm {
  def kruskal[VD:scala.reflect.ClassTag](g: Graph[VD , Double]): Graph[VD, Double] ={

    //将边改为二维元组 其中 _1为边的属性
    var g2 = g.mapEdges(e => (e.attr, false))

    for(i<-1L to g.vertices.count()-1){

      val miniTreeVertices = g2.subgraph(_.attr._2) //选择已经被选择的边
        .connectedComponents()                      //获取联通组件(此时联通组件有且只有一个 因为最小树的必是联通的)
        .vertices                                   //获取顶点
                                                    //Note 这里顶点返回的是个二元元组
                                                    //其中_1对应的是顶点
                                                    //_2 对应的是该顶点对应的联通组件id(用组件内最小的顶点表示)

      val unavailableEdges = g2.outerJoinVertices(miniTreeVertices)((vid,vd,cid)=>(vd,cid)) //cid 为联通组件id
        .subgraph(et=>et.srcAttr._2.getOrElse(-1) == et.dstAttr._2.getOrElse(-2)) //当边都属于最小树联通组件的话，
                                                                                  //该边是无效的 因为加入后会导致树成环
        .edges
        .map(e=>((e.srcId,e.dstId),e.attr))

      type edgeType = ((VertexId,VertexId),Double)

      //获取当前最小的边
      //这里只是查找最小边对方向未做约束
      val smallestEdge = g2.edges
        .map(e=>((e.srcId,e.dstId),e.attr))
        .leftOuterJoin(unavailableEdges)           //最终的join后的结果是 ((srcId,dstId),((权重,是否已在最小树),SOME(权重,是否已在最小树)))
        .filter(x=> !x._2._1._2 && x._2._2.isEmpty) //不在最小树内其不是无效边
        .map(x=>(x._1,x._2._1._1))      // 二元元组 ((srcId,dstId),权重)
        .min()(new Ordering[edgeType](){
        override def compare(a: edgeType, b: edgeType):Int = {
          val r =Ordering[Double].compare(a._2,b._2) //判断权重
          if (r==0)
            Ordering[Long].compare(a._1._1,b._1._1) //权重相同选节点编号小的
          else
            r
        }
      })

      //将权重最小的边 归入最小树
      g2=g2.mapTriplets(et=>(et.attr._1,et.attr._2 || (et.srcId==smallestEdge._1._1 && et.dstId==smallestEdge._1._2)))
    }

    g2.subgraph(_.attr._2).mapEdges(_.attr._1)
  }
}
